package netty;

import com.bjtu.boatmq.common.protocol.*;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.util.Date;

class Student{
    private String name;
    private int age;
    private String sex;

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public int getAge() {
        return age;
    }

    public void setAge(int age) {
        this.age = age;
    }

    public String getSex() {
        return sex;
    }

    public void setSex(String sex) {
        this.sex = sex;
    }

    public Student(String name, int age, String sex){
        name = name;
        age = age;
        sex = sex;
    }
}

public class NettyClient {
    public static void main(String[] args) throws InterruptedException {
        Bootstrap bootstrap = new Bootstrap();
        NioEventLoopGroup group = new NioEventLoopGroup();

        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel ch) {
                        ch.pipeline().addLast(new SourceEncoder());
                        ch.pipeline().addLast(new SourceDecoder());
                        ch.pipeline().addLast(new SimpleChannelInboundHandler<PacketSource>() {
                            @Override
                            protected void channelRead0(ChannelHandlerContext ctx, PacketSource ps) {
                                System.out.println("接收到一条broker返回的消息["+ps.getRequestId()+"]:"+ps.getRemark());
                                System.out.println("offset:"+ps.getRemark());
                                // List<Message> messageList = (ArrayList<Message>)ps.getData();
                                // System.out.println("msg size:"+messageList.toString());
                            }
                        });
                    }
                });

        ChannelFuture future = bootstrap.connect("127.0.0.1", 10110);
        // 拉取消息
        /*new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    PullMessageRequest pullMessageRequest = new PullMessageRequest();
                    pullMessageRequest.setTopicName("topic-0");
                    pullMessageRequest.setPartitionId(0);
                    pullMessageRequest.setConsumeOffset(PullMessageRequest.OFFSET_START);

                    PacketSource packetSource = new PacketSource();
                    packetSource.setCode(RequestCode.PULL_MESSAGE);
                    packetSource.setSponsorType(PacketSource.CONSUMER);
                    packetSource.setSponsorId("consumer001");
                    packetSource.setData(pullMessageRequest);  //

                    future.channel().writeAndFlush(packetSource);
                    System.out.println("拉取消息:"+packetSource.getRequestId());
                    try {
                        Thread.sleep(1000);
                    }catch (InterruptedException e){
                        e.printStackTrace();
                    }
                }
            }
        }).start();*/
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    PacketSource packetSource = new PacketSource();
                    packetSource.setCode(RequestCode.SEND_MESSAGE);
                    packetSource.setSponsorType(PacketSource.PRODUCER);

                    //List<Message> messageList = new ArrayList<>();
                    Message message = new Message();
                    message.setTopic("topic-0");
                    message.setPartitionId(0);
                    message.setBody(new Date() + ": this is a message! ["+packetSource.getRequestId()+"]");
                    // messageList.add(message);

                    packetSource.setData(message);
                    System.out.println("data type:"+packetSource.getData().getClass().getName());
                    future.channel().writeAndFlush(packetSource);
                    System.out.println("future:"+future.channel().read());
                    System.out.println("【分区0】发送消息："+packetSource.getRequestId());
                    try {
                        Thread.sleep(1000);
                    }catch (InterruptedException e){
                        e.printStackTrace();
                    }
                }
            }
        }).start();
       /* new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    PacketSource packetSource = new PacketSource();
                    packetSource.setCode(RequestCode.SEND_MESSAGE);
                    packetSource.setSponsorType(PacketSource.CONSUMER);
                    packetSource.setSponsorId("consumer001");
                    Message message = new Message();
                    message.setTopic("topic-0");
                    message.setPartitionId(1);
                    message.setBody(new Date() + ": this is a message! ["+packetSource.getRequestId()+"]");
                    packetSource.setData(message);
                    future.channel().writeAndFlush(packetSource);
                    System.out.println("future:"+future.channel().read());
                    System.out.println("【分区1】发送消息："+packetSource.getRequestId());
                    try {
                        Thread.sleep(1000);
                    }catch (InterruptedException e){
                        e.printStackTrace();
                    }
                }
            }
        }).start();*/


    }
}